// Copyright (c) 2019 Presto Labs Pte. Ltd.
// Author: jaewon

#include "coin1/fastfeed/feed_cache.h"

#include <functional>
#include <string>

#include <glog/logging.h>
#include <gtest/gtest.h>

#include "coin1/fastfeed/view/string_json_builder.h"

namespace fastfeed {

void DoNothing(const proto::Feed& feed) {}

TEST(FEED_CACHE_READER, BASICS) {
  FeedCacheReader fcreader("20190501", {"/home/jaewon/data/feed_cache"});

  fcreader.AddCacheKey(
      FeedCacheKey {"feed-01.ap-northeast-1.aws", "1", "Futures.Bitmex"},
      {"BTC-USD.PERPETUAL.Bitmex.XBTUSD", "ETH-USD.PERPETUAL.Bitmex.ETHUSD"},
      DoNothing);

  for (int i = 0; i < 10; ++i) {
    auto res = fcreader.Read();
    if (!res) break;
  }
}

class Handler {
 public:
  Handler(const FeedCacheKey& feed_cache_key, const std::vector<std::string>& symbols)
      : feed_cache_key_(feed_cache_key), symbols_(symbols) {}

  void OnFeedBuilderReset(const FeedCacheKey& key, FeedBuilderBase* feed_builder) {
    CHECK(feed_cache_key_ == key);
    LOG(INFO) << "FeedBuilder reset: " << key.ToString();
    auto book_cb = std::bind(&Handler::OnBook, this, std::placeholders::_1);
    auto trade_cb = std::bind(&Handler::OnTrade, this, std::placeholders::_1);
    auto inst_cb = std::bind(&Handler::OnInstrument, this, std::placeholders::_1);
    for (auto&& symbol : symbols_) {
      feed_builder->SubscribeBook(symbol, book_cb);
      feed_builder->SubscribeTrade(symbol, trade_cb);
      feed_builder->Subscribe<view::StringJsonBuilder>(
          proto::FEED_TYPE_CUSTOM,
          "bitmex_instrument",
          symbol,
          "json",
          inst_cb);
    }
  }

  int book_count() const { return book_count_; }
  int trade_count() const { return trade_count_; }
  int inst_count() const { return inst_count_; }

 private:
  void OnBook(const view::BookBuilder* book) {
    ++book_count_;
    LOG_EVERY_N(INFO, 1000000) << "book " << book_count_;
  }

  void OnTrade(const view::TradeBuilder* trade) {
    ++trade_count_;
    LOG_EVERY_N(INFO, 100000) << "trade " << trade_count_;
  }

  void OnInstrument(const view::StringJsonBuilder* builder) {
    ++inst_count_;
    LOG_EVERY_N(INFO, 100000) << "instrument " << inst_count_;
  }

  int book_count_ = 0;
  int trade_count_ = 0;
  int inst_count_ = 0;

  FeedCacheKey feed_cache_key_;
  std::vector<std::string> symbols_;
};  // class Handler

TEST(FEED_CACHE_RUNNER, BASICS) {
  FeedCacheKey cache_key {"feed-01.ap-northeast-1.aws", "", "Futures.Bitmex"};
  std::vector<std::string> symbols {
      "BTC-USD.PERPETUAL.Bitmex.XBTUSD",
      "ETH-USD.PERPETUAL.Bitmex.ETHUSD"};

  Handler handler(cache_key, symbols);
  auto reset_callback = std::bind(
      &Handler::OnFeedBuilderReset,
      &handler,
      std::placeholders::_1,
      std::placeholders::_2);
  FeedCacheRunner runner(reset_callback, "20190401", {"/home/jaewon/data/feed_cache"});

  runner.AddCacheKeyWithFeedArb(cache_key.machine, cache_key.sub_req, symbols);
  runner.Run();

  LOG(INFO) << "# book messages: " << handler.book_count();
  LOG(INFO) << "# trade messages: " << handler.trade_count();
  LOG(INFO) << "# instrument messages: " << handler.inst_count();
}

}  // namespace fastfeed
